package com.liser.southernstar.utils;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Scheduler;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;

/**
 * 封装RxJava2模型为一次性任务。<br/>
 * 在此模型中，被订阅者执行完毕后将立即释放订阅资源
 * <p style='color:red;'>在RxJava2中调用onError会关闭订阅资源，两次调用会发生严重错误。为保证程序健壮性如需触发onError请直接抛出异常</p>
 * @param <T>
 */
public class RxJava2Task<T>{
    private ObservableOnSubscribe<T> source;
    private Scheduler scheduler;
    private Disposable disposable = null;

    public RxJava2Task(ObservableOnSubscribe<T> source, Scheduler scheduler){
        this.source = source;
        this.scheduler = scheduler;
    }

    public void subscribeOnce(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete){
        if(disposable != null){
            throw new IllegalStateException("此任务实例已在运行中或已被使用");
        }
        // 被订阅内容
        Observable<T> observable = Observable.create((ObservableEmitter<T> emitter) -> {
            try{
                source.subscribe(emitter);
            }catch (Exception ex){
                // onError调用一次即被关闭
                if(!isDisposed()){
                    emitter.onError(ex);
                }
            }finally {
                release();
            }
        });
        // 订阅
        if(onNext != null && onError != null && onComplete != null){
            disposable = observable.subscribeOn(scheduler).subscribe(onNext, onError, onComplete);
        }else if(onNext != null && onError != null){
            disposable = observable.subscribeOn(scheduler).subscribe(onNext, onError);
        }else if(onNext != null){
            disposable = observable.subscribeOn(scheduler).subscribe(onNext);
        }else{
            disposable = observable.subscribeOn(scheduler).subscribe();
        }
    }

    public void subscribeOnce(Consumer<? super T> onNext, Consumer<? super Throwable> onError){
        this.subscribeOnce(onNext, onError, null);
    }

    public void subscribeOnce(Consumer<? super T> onNext){
        this.subscribeOnce(onNext, null, null);
    }

    public void subscribeOnce(){
        this.subscribeOnce(null, null, null);
    }

    public boolean isDisposed(){
        return disposable == null || disposable.isDisposed();
    }

    /**释放资源*/
    private void release(){
        if(!isDisposed()){
            disposable.dispose();
        }
    }
}
